package com.itat.client;

import com.alibaba.fastjson.JSON;
import com.itat.protocol.RpcDecoder;
import com.itat.protocol.RpcEncoder;
import com.itat.protocol.RpcRequest;
import com.itat.protocol.RpcResponse;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import java.util.Date;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.TimeUnit;


/**
 * 重试连接板
 * netty client
 * @author baixuefeng
 * @date 2021-12-29
 */

public class ReconnectClient {

    private static final Log log = LogFactory.getLog(ReconnectClient.class);

    //默认端口
    private static final int defaultPort=56709;
    //重试连接时间
    private final int retryConnectTime = 10;//s



    private NioEventLoopGroup workGroup = new NioEventLoopGroup(4);
    private Bootstrap bootstrap=null;
    //重试队列
    private static final ArrayBlockingQueue<RpcRequest> RETRYSENDQUENE=new ArrayBlockingQueue<RpcRequest>(1000);

    //ip和端口
    private final String host;
    private final int port;
    private Channel channel;

    //连接服务端的端口号地址和端口号
    public ReconnectClient(String host, int port) {
        this.host = host;
        this.port = port;
    }

    /**
     * 客户端直接发送数据
     * @param request
     * @throws Exception
     */
    public void sendData(RpcRequest request) {
        try{
            if (channel != null && channel.isActive()) {
                System.out.println("netty客户端发送数据:"+ JSON.toJSONString(request));
                //channel对象可保存在map中，供其它地方发送消息
                ChannelFuture flag=channel.writeAndFlush(request);
                if(!flag.isSuccess()){
                    //重新放进去
                    RETRYSENDQUENE.put(request);
                    return;
                }
            }else{
                //
                RETRYSENDQUENE.put(request);
                return;
            }
        } catch (Exception e) {
            log.warn("netty客户端发送数据失败",e);
        }
    }

    /**
     * 客户端从队列Quene取数据发送
     * 所有数据统一Quene发送
     * 客户端消息重试机制: 数据重发
     */
    protected void retrySendDataMonitor() {
        System.out.println("retry loop start:");
        while(true){
            try{
//                log.warn("retry loop doing:"+ new Date());
                //channel对象可保存在map中，供其它地方发送消息
                if (channel != null && channel.isActive()) {
                    //线程阻塞
                    RpcRequest  request= RETRYSENDQUENE.take();
                    System.out.println("netty客户端retry发送:"+ JSON.toJSONString(request));
                    ChannelFuture flag= channel.writeAndFlush(request);
                    if(!flag.isSuccess()){
                        //重新放进去
                        RETRYSENDQUENE.put(request);
                        return;
                    }
                }
                Thread.sleep(1000);
            } catch (Exception e) {
                log.warn("netty客户端retry发送数据失败",e);
            }
        }
    }


    //客户端启动
    public void start() {
//        final EventLoopGroup workGroup = new NioEventLoopGroup();//单独申明Group
        try {
            bootstrap = new Bootstrap();
            bootstrap
                    .group(workGroup)
                    .channel(NioSocketChannel.class)
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            System.out.println("正在连接中...");
                            ChannelPipeline pipeline = ch.pipeline();
                            //发送数据
                            pipeline.addLast(new RpcEncoder(RpcRequest.class)); //编码request
                            pipeline.addLast(new RpcDecoder(RpcResponse.class)); //解码response
                            pipeline.addLast(new ReconnectClientHandler(ReconnectClient.this));

                        }
                    });
            doConnect(retryConnectTime);
        } catch (Exception e) {
            log.warn("netty客户端初始化连接失败",e);
        }

        //消息重试机制
        new Thread(new Runnable() {
            public void run() {
                //
                retrySendDataMonitor();
            }
        }).start();

    }

    //客户端连接
    protected void doConnect(final long retrySecondTime)  {
        if (channel != null && channel.isActive()) {
            return;
        }
//
        ChannelFuture future=null;
        try {
            future = bootstrap.connect(host, port);

            future.addListener(new ChannelFutureListener() {
                public void operationComplete(ChannelFuture futureListener)  {
                    if (futureListener.isSuccess()) {
                        channel = futureListener.channel();
                        System.out.println("Connect to server successfully!");
                        //连接成功，线程解锁
                    } else {
                        log.error("Failed to connect to server, try to connect after 10s");
                        futureListener.channel().eventLoop().schedule(new Runnable() {
                            public void run() {
                                try {
                                    doConnect(retrySecondTime);
                                } catch (Exception e) {
                                    log.warn("netty客户端连接失败",e);
                                }
                            }
                        }, retrySecondTime, TimeUnit.SECONDS);
                    }
                }
            });
            future.sync();/** 此处连接服务器失败会抛出异常 **/
            this.channel = future.channel();
        } catch (Exception e) {
            log.warn("netty客户端连接失败",e);
        }finally {
            if (RETRYSENDQUENE.isEmpty()) {//没有数据
                return;
            }
        }

    }

    public Channel getChannel() {
        return channel;
    }

    //消息加入队列
    public static  void addQuene(RpcRequest request){
        try {
            RETRYSENDQUENE.put(request);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }



}
